package com.fancoder.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class ProducerBootstrap {

    private String brokers = "192.168.121.80:9092";
    private String topic = "topic-demo";

    public static void main(String[] args) {
        ProducerBootstrap producerBootstrap = new ProducerBootstrap();
        producerBootstrap.start();
    }

    private void start() {
        Properties properties = initConfig();
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        List<PartitionInfo> partitionInfos = producer.partitionsFor(topic);
        ProducerRecord<String, String> messageNoKey = new ProducerRecord<String, String>(topic, "message body 1");
        Future<RecordMetadata> send = producer.send(messageNoKey);
        try {
            System.out.println(send.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        ProducerRecord<String, String> messageWithKey = new ProducerRecord<String, String>(topic, "message body 1");
        Future<RecordMetadata> send2 = producer.send(messageWithKey);
        try {
            System.out.println(send2.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

    private Properties initConfig() {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-client-id-01");
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, ProducerPartitioner.class.getName());
        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerInterceptor.class.getName());
        return properties;
    }
}
